package com.gukun.springboot.tdengine.service.writeplant;

import com.gukun.springboot.tdengine.domain.enums.TagType;
import com.zaxxer.hikari.HikariDataSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.HashMap;
import java.util.Map;

/**
 * A helper class encapsulate the logic of writing using SQL.
 * <p>
 * The main interfaces are two methods:
 * <ol>
 *     <li>{@link SQLWriter#processLine}, which receive raw lines from WriteTask and group them by table names.</li>
 *     <li>{@link  SQLWriter#flush}, which assemble INSERT statement and execute it.</li>
 * </ol>
 * <p>
 * There is a technical skill worth mentioning: we create table as needed when "table does not exist" error occur instead of creating table automatically using syntax "INSET INTO tb USING stb".
 * This ensure that checking table existence is a one-time-only operation.
 * </p>
 *
 * </p>
 */
public class SQLWriter {
    final static Logger logger = LoggerFactory.getLogger(SQLWriter.class);

    private Map<String, HikariDataSource> hdMap = new HashMap<>();

    private Connection conn;
    private Statement stmt;

    /**
     * current number of buffered records
     */
    private int bufferedCount = 0;
    /**
     * Maximum number of buffered records.
     * Flush action will be triggered if bufferedCount reached this value,
     */
    private int maxBatchSize;


    /**
     * Maximum SQL length.
     */
    private int maxSQLLength = 1000000;

    private String jdbcURL;

    private String dbName;

    /**
     * Map from table name to column values. For example:
     * "tb001" -> "(1648432611249,2.1,114,0.09) (1648432611250,2.2,135,0.2)"
     */
    private Map<String, String> tbValues = new HashMap<>();

    /**
     * Map from table name to column types. For example:
     * "tb001" -> "FLOAT","tb002" -> "DOUBLE"
     */
    private Map<String, String> tbTypes = new HashMap<>();

    /**
     * Map from table name to tag values in the same order as creating stable.
     * Used for creating table.
     */
    private Map<String, String> tbTags = new HashMap<>();

    public SQLWriter(int maxBatchSize,String jdbcURL,String dbName) {
        this.maxBatchSize = maxBatchSize;
        this.jdbcURL = jdbcURL;
        this.dbName = dbName;
    }


    /**
     * Get Database Connection
     *
     * @return Connection
     * @throws SQLException
     */
    private Connection getConnection() throws SQLException {
//        String jdbcURL = "jdbc:TAOS://10.88.12.71:6030?charset=UTF-8&timezone=UTC-8&user=root&password=taosdata";
        return getConnectionPool(jdbcURL,2);
    }

    public Connection getConnectionPool(String url,int poolsize) throws SQLException {
        if(hdMap.containsKey(url)){
            HikariDataSource hikariDataSource = hdMap.get(url);
            Connection connection = hikariDataSource.getConnection();
            if(!connection.isValid(1000) || connection.isClosed()){
                hikariDataSource = TDEngineConnPool.getHikariDataSource(url,poolsize);
                connection = hikariDataSource.getConnection();
                hdMap.put(url, hikariDataSource);
            }
            return connection;
        }else{
            HikariDataSource hikariDataSource = TDEngineConnPool.getHikariDataSource(url,poolsize);
            Connection connection = hikariDataSource.getConnection();
            hdMap.put(url, hikariDataSource);
            return connection;
        }
    }

    /**
     * Create Connection and Statement
     *
     * @throws SQLException
     */
    public void init() throws SQLException {
        conn = getConnection();
        stmt = conn.createStatement();
        stmt.execute("use "+dbName);
    }

    /**
     * Convert raw data to SQL fragments, group them by table name and cache them in a HashMap.
     * Trigger writing when number of buffered records reached maxBachSize.
     *
     * @param line raw data get from task queue in format: tbName,ts,current,voltage,phase,location,groupId
     */
    public void processLine(String line) throws SQLException {

        bufferedCount += 1;
        int firstComma = line.indexOf(',');
        String tbName = line.substring(0, firstComma);

        //String value = "(" + line.substring(firstComma + 1) + ") ";
        //value原本的值：时间，值，质量，此处需要将质量的值给去掉
        String oldValue = line.substring(firstComma + 1);
        String[] arr = oldValue.split(",");
        if(arr.length == 3){
            String value = "(" +arr[0]+","+arr[1]  + ") ";
            if (tbValues.containsKey(tbName)) {
                tbValues.put(tbName, tbValues.get(tbName) + value);
            } else {
                tbValues.put(tbName, value);
            }
//        if (!tbTags.containsKey(tbName)) {
//            String tagValues =  "(" + tbName.split("_")[1]+ ')';
//            tbTags.put(tbName, tagValues);
//        }
            if (bufferedCount == maxBatchSize) {
                flush();
            }
        }
    }


    /**
     * Assemble INSERT statement using buffered SQL fragments in Map {@link SQLWriter#tbValues} and execute it.
     * In case of "Table does not exit" exception, create all tables in the sql and retry the sql.
     */
    public void flush() throws SQLException {
        StringBuilder sb = new StringBuilder("INSERT INTO ");
        for (Map.Entry<String, String> entry : tbValues.entrySet()) {
            String tableName = entry.getKey();
            String values = entry.getValue();
            String q = tableName + " values " + values + " ";
            if (sb.length() + q.length() > maxSQLLength) {
                executeSQL(sb.toString());
//                logger.warn("increase maxSQLLength or decrease maxBatchSize to gain better performance");
                sb = new StringBuilder("INSERT INTO ");
            }
            sb.append(q);
        }
        executeSQL(sb.toString());
        tbValues.clear();
        bufferedCount = 0;
    }

    private void executeSQL(String sql) throws SQLException {
        try {
            stmt.executeUpdate(sql);
        } catch (SQLException e) {
            // convert to error code defined in taoserror.h
            int errorCode = e.getErrorCode() & 0xffff;
            if (errorCode == 0x2603) {
                // Table does not exist
                createTables();
                executeSQL(sql);
            } else {
                logger.error("Execute SQL: {}", sql);
                throw e;
            }
        } catch (Throwable throwable) {
            logger.error("Execute SQL: {}", sql);
            throw throwable;
        }
    }

    /**
     * Create tables in batch using syntax:
     * <p>
     * CREATE TABLE [IF NOT EXISTS] tb_name1 USING stb_name TAGS (tag_value1, ...) [IF NOT EXISTS] tb_name2 USING stb_name TAGS (tag_value2, ...) ...;
     * </p>
     */
    private void createTables() throws SQLException {
        StringBuilder sb = new StringBuilder("CREATE TABLE ");
        int index = 0;
        for (String tbName : tbValues.keySet()) {
//            String tagValues = tbTags.get(tbName);
            //String superTableName = getSTableName(tbTypes.get(tbName));
            String superTableName = getSTable(tbName);
            sb.append("IF NOT EXISTS ").append(tbName).append(" USING "+superTableName+" TAGS (")
                    .append("'"+tbName+"'") //ti的值
                    .append(",")
                    .append("'notn',")
                    .append("'1'")
                    .append(") ");
            index++;
            if(index == 4000){
                String sql = sb.toString();
                try {
                    stmt.executeUpdate(sql);
                } catch (Throwable throwable) {
                    logger.error("Execute SQL: {}", sql);
                    throw throwable;
                }
                sb = new StringBuilder("CREATE TABLE ");
                index = 0;
            }
        }
        String sql = sb.toString();
        if(!"CREATE TABLE ".equals(sql)){
            try {
                stmt.executeUpdate(sql);
            } catch (Throwable throwable) {
                logger.error("Execute SQL: {}", sql);
                throw throwable;
            }
        }
    }

    private String getSTable(String tbName) {
        String stag = "tfloat";
        if(tbName != null){
            String val = tbName.toLowerCase();
            if(val.indexOf("float")>-1){
                stag = "tfloat";
            }else if(val.indexOf("double")>-1){
                stag = "tdouble";
            }else if(val.indexOf("int")>-1){
                stag = "tint";
            }else if(val.indexOf("bool")>-1){
                stag = "tbool";
            }
        }
        return stag;
    }

    /**
     * 获取超表名称
     *
     * @param type
     * @return
     */
    private String getSTableName(String type) {
        if(type == null){
            return "tfloat";
        }
        String stag = "t" + type.toLowerCase();
        switch (type) {
            case "DOUBLE":
            case "BOOL":
            case "INT":
            case "FLOAT":
                stag = "t" + type.toLowerCase();
                break;
        }
        return stag;
    }

    public boolean hasBufferedValues() {
        return bufferedCount > 0;
    }

    public int getBufferedCount() {
        return bufferedCount;
    }

    public void close() {
        try {
            stmt.close();
        } catch (SQLException e) {
        }
        try {
            conn.close();
        } catch (SQLException e) {
        }
    }
}

